package com.example.rocketmq.demo.controller;

import com.example.rocketmq.demo.config.RocketMQConstant;
import com.example.rocketmq.demo.producer.AdvancedProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 高级消息控制器
 * 提供同步和异步消息的更多场景
 */
@Slf4j
@RestController
@RequestMapping("/rocketmq/advanced")
public class AdvancedMsgController {

    @Autowired
    private AdvancedProducer advancedProducer;

    /**
     * 同步发送消息并设置超时时间
     */
    @GetMapping("/sync/timeout")
    public Map<String, Object> sendSyncWithTimeout(
            @RequestParam String message,
            @RequestParam(defaultValue = "3000") long timeout) {
        Map<String, Object> result = new HashMap<>();
        
        try {
            SendResult sendResult = advancedProducer.sendSyncWithTimeout(
                    RocketMQConstant.TOPIC_NORMAL,
                    RocketMQConstant.TAG_NORMAL,
                    "key-" + System.currentTimeMillis(),
                    message,
                    timeout
            );
            
            result.put("success", true);
            result.put("msgId", sendResult.getMsgId());
            result.put("queueId", sendResult.getMessageQueue().getQueueId());
            result.put("brokerName", sendResult.getMessageQueue().getBrokerName());
            result.put("sendStatus", sendResult.getSendStatus().name());
        } catch (Exception e) {
            log.error("同步发送消息失败", e);
            result.put("success", false);
            result.put("error", e.getMessage());
        }
        
        return result;
    }

    /**
     * 异步发送消息并等待结果
     */
    @GetMapping("/async/wait")
    public Map<String, Object> sendAsyncAndWait(
            @RequestParam String message,
            @RequestParam(defaultValue = "5") int waitSeconds) {
        Map<String, Object> result = new HashMap<>();
        
        try {
            CompletableFuture<SendResult> future = advancedProducer.sendAsyncWithFuture(
                    RocketMQConstant.TOPIC_NORMAL,
                    RocketMQConstant.TAG_NORMAL,
                    "key-" + System.currentTimeMillis(),
                    message
            );
            
            // 等待异步结果
            SendResult sendResult = future.get(waitSeconds, TimeUnit.SECONDS);
            
            result.put("success", true);
            result.put("msgId", sendResult.getMsgId());
            result.put("queueId", sendResult.getMessageQueue().getQueueId());
            result.put("brokerName", sendResult.getMessageQueue().getBrokerName());
            result.put("sendStatus", sendResult.getSendStatus().name());
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("异步发送消息失败或等待超时", e);
            result.put("success", false);
            result.put("error", e.getMessage());
        }
        
        return result;
    }

    /**
     * 异步发送多条消息
     */
    @PostMapping("/async/multiple")
    public Map<String, Object> sendMultipleAsync(@RequestBody String[] messages) {
        Map<String, Object> result = new HashMap<>();
        
        try {
            boolean allSuccess = advancedProducer.sendMultipleAsync(
                    RocketMQConstant.TOPIC_NORMAL,
                    RocketMQConstant.TAG_NORMAL,
                    messages
            );
            
            result.put("success", allSuccess);
            result.put("messageCount", messages.length);
            result.put("messages", Arrays.asList(messages));
        } catch (Exception e) {
            log.error("批量异步发送消息失败", e);
            result.put("success", false);
            result.put("error", e.getMessage());
        }
        
        return result;
    }

    /**
     * 同步发送消息并等待消费确认
     */
    @GetMapping("/sync/await-consumption")
    public Map<String, Object> sendAndAwaitConsumption(
            @RequestParam String message,
            @RequestParam(defaultValue = "5") int timeout) {
        Map<String, Object> result = new HashMap<>();
        
        try {
            String response = advancedProducer.sendAndAwaitConsumption(message, timeout);
            
            result.put("success", true);
            result.put("response", response);
        } catch (Exception e) {
            log.error("发送并等待消费确认失败", e);
            result.put("success", false);
            result.put("error", e.getMessage());
        }
        
        return result;
    }
}